package com.lu.integrate.message.center.service;
import java.util.ArrayList;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.lu.integrate.message.center.model.enums.MessageQueueLogsStatusEnum;
import com.lu.stack.common.model.response.R;
import com.lu.integrate.message.center.model.domain.MessageCenterQueueConsumeDo;
import com.lu.integrate.message.center.model.domain.MessageCenterQueueMessageDo;
import com.lu.integrate.message.center.model.request.MessageCenterQueueMessageConfirmSubmitRequest;
import com.lu.integrate.message.center.model.request.MessageCenterQueueProduceRequest;
import com.lu.integrate.message.center.model.entity.SysMessageQueueConfig;
import com.lu.integrate.message.center.model.entity.SysMessageQueueLogs;
import com.lu.integrate.message.center.model.request.MessageCenterQueueRetryConsumerRequest;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author zhanglu
 * @project lu-cloud-stack
 * @description 基于mysql的消息队列
 * @date 2023/2/13 17:48
 */
@Slf4j
@Service
public class DbMessageQueueService extends MessageQueueAbstract implements MessageQueueService{

    @Resource
    private SysMessageQueueConfigService messageQueueConfigService;
    @Resource
    private SysMessageQueueLogsService messageQueueLogsService;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public R produce(MessageCenterQueueProduceRequest produceDo) {
        QueryWrapper<SysMessageQueueConfig> configQueryWrapper = new QueryWrapper<>();
        configQueryWrapper.eq("topic", produceDo.getTopic())
                .eq("class_name", produceDo.getClassName())
                .eq("method_name", produceDo.getMethodName());
        configQueryWrapper.last("order by id desc limit 1");
        SysMessageQueueConfig config = messageQueueConfigService.getOne(configQueryWrapper);
        if(config == null){
            SysMessageQueueConfig saveConfig = new SysMessageQueueConfig();
            saveConfig.setTopic(produceDo.getTopic());
            saveConfig.setClassName(produceDo.getClassName());
            saveConfig.setMethodName(produceDo.getMethodName());
            saveConfig.setCreateBy("lu");
            messageQueueConfigService.save(saveConfig);
            config = saveConfig;
        }
        SysMessageQueueLogs logs = new SysMessageQueueLogs();
        logs.setConfigId(config.getId());
        logs.setMessage(produceDo.getMessage());
        messageQueueLogsService.save(logs);
        return R.success().params("config_id", config.getId()).params("log_id", logs.getId());
    }

    @Override
    public R consume(MessageCenterQueueConsumeDo consumeDo) {
        if(consumeDo.getStatus() == MessageQueueLogsStatusEnum.WAIT.getStatus()){
            return waitConsume(consumeDo);
        }else if(consumeDo.getStatus() == MessageQueueLogsStatusEnum.PROCESSING.getStatus()){
            return processingConsume(consumeDo);
        }
        return R.success();
    }

    @Override
    public R confirmSubmit(MessageCenterQueueMessageConfirmSubmitRequest request) {
        SysMessageQueueLogs logs = new SysMessageQueueLogs();
        logs.setId(request.getLogId());
        logs.setStatus(request.getStatus());
        logs.setEndExecuteTime(DateUtil.date());
        logs.setExecuteResult(request.getResult());
        messageQueueLogsService.updateById(logs);
        return R.success(messageQueueLogsService.getById(logs.getId()));
    }

    @Override
    public R retryConsume(MessageCenterQueueRetryConsumerRequest request) {
        SysMessageQueueConfig config = messageQueueConfigService.getById(request.getConfigId());
        //查询topic的消息
        QueryWrapper<SysMessageQueueLogs> queryWrapper = new QueryWrapper();
        queryWrapper.select("id,config_id,status,message");
        queryWrapper.eq("config_id", request.getConfigId());
        if(ObjectUtil.isNotEmpty(request.getLogsIds())){
            queryWrapper.in("id", request.getLogsIds());
        }
        List<SysMessageQueueLogs> logs = messageQueueLogsService.list(queryWrapper);
        logs.forEach(queueLog -> {
            //反射调用方法，传递logId
            log.info("db消息队列执行器-处理中消息消费,id:{}---消息:{}", queueLog.getId(), queueLog.getMessage());
            R r = invokeConsumer(config.getClassName(), config.getMethodName(), () -> {
                MessageCenterQueueMessageDo messageDo = new MessageCenterQueueMessageDo();
                messageDo.setLogId(queueLog.getId());
                messageDo.setMessage(queueLog.getMessage());
                return messageDo;
            });
            log.info("db消息队列执行器-处理中消息消费,id:{}---结果:{}", queueLog.getId(), JSON.toJSONString(r));
        });
        return R.success(logs);
    }

    /**
     * 待处理 消息消费
     * @param consumeDo
     * @return
     */
    private R waitConsume(MessageCenterQueueConsumeDo consumeDo){
        List<R> rs = new ArrayList<>();
        if(log.isDebugEnabled()){
            log.debug("db消息队列执行器-待处理消息消费,方法请求参数:{}", JSON.toJSONString(consumeDo));
        }
        //查询topic的消息
        QueryWrapper<SysMessageQueueLogs> queryWrapper = new QueryWrapper();
        queryWrapper.select("id,config_id,status,message");
        if(ObjectUtil.isNotEmpty(consumeDo.getConfigId())){
            queryWrapper.eq("config_id", consumeDo.getConfigId());
        }
        queryWrapper.eq("status", MessageQueueLogsStatusEnum.WAIT.getStatus());
        queryWrapper.last("limit "+consumeDo.getBatchSize()+"");
        List<SysMessageQueueLogs> logs = messageQueueLogsService.list(queryWrapper);
        log.info("db消息队列执行器-待处理消息消费,消息量:{}", logs.size());
        logs.forEach(queueLog -> {
            //更新消息状态
            UpdateWrapper<SysMessageQueueLogs> updateWrapper = new UpdateWrapper<>();
            updateWrapper.eq("id", queueLog.getId());
            updateWrapper.eq("status", MessageQueueLogsStatusEnum.WAIT.getStatus());
            SysMessageQueueLogs updateLogs = new SysMessageQueueLogs();
            updateLogs.setStatus(MessageQueueLogsStatusEnum.PROCESSING.getStatus());
            updateLogs.setStartExecuteTime(DateUtil.date());
            updateLogs.setExecuteResult("waitConsume");
            boolean update = messageQueueLogsService.update(updateLogs, updateWrapper);
            if(!update){
                log.info("db消息队列执行器-待处理消息消费存在并发问题,更新消息状态失败,id:{}---状态:{}", queueLog.getId(), queueLog.getStatus());
                return;
            }
            //反射调用方法，传递logId
            log.info("db消息队列执行器-待处理消息消费,id:{}---消息:{}", queueLog.getId(), queueLog.getMessage());
            R r = invokeConsumer(consumeDo.getClassName(), consumeDo.getMethodName(), () -> {
                MessageCenterQueueMessageDo messageDo = new MessageCenterQueueMessageDo();
                messageDo.setLogId(queueLog.getId());
                messageDo.setMessage(queueLog.getMessage());
                return messageDo;
            });
            rs.add(r);
            log.info("db消息队列执行器-待处理消息消费,id:{}---结果:{}", queueLog.getId(), JSON.toJSONString(r));
        });
        return R.success(rs);
    }

    /**
     * 处理中 消息消费
     * @param consumeDo
     * @return
     */
    private R processingConsume(MessageCenterQueueConsumeDo consumeDo) {
        List<R> rs = new ArrayList<>();
        if(log.isDebugEnabled()){
            log.debug("db消息队列执行器-处理中消息消费,方法请求参数:{}", JSON.toJSONString(consumeDo));
        }
        //查询topic的消息
        QueryWrapper<SysMessageQueueLogs> queryWrapper = new QueryWrapper();
        queryWrapper.select("id,config_id,status,message");
        if(ObjectUtil.isNotEmpty(consumeDo.getConfigId())){
            queryWrapper.eq("config_id", consumeDo.getConfigId());
        }
        queryWrapper.eq("status", MessageQueueLogsStatusEnum.PROCESSING.getStatus());
        queryWrapper.last("limit "+consumeDo.getBatchSize()+"");
        List<SysMessageQueueLogs> logs = messageQueueLogsService.list(queryWrapper);
        log.info("db消息队列执行器-处理中消息消费,消息量:{}", logs.size());
        logs.forEach(queueLog -> {
            //更新消息状态
            UpdateWrapper<SysMessageQueueLogs> updateWrapper = new UpdateWrapper<>();
            updateWrapper.eq("id", queueLog.getId());
            updateWrapper.eq("status", MessageQueueLogsStatusEnum.PROCESSING.getStatus());
            SysMessageQueueLogs updateLogs = new SysMessageQueueLogs();
            updateLogs.setStartExecuteTime(DateUtil.date());
            updateLogs.setExecuteResult("processingConsume");
            boolean update = messageQueueLogsService.update(updateLogs, updateWrapper);
            if(!update){
                log.info("db消息队列执行器-处理中消息消费存在并发问题,更新消息状态失败,id:{}---状态:{}", queueLog.getId(), queueLog.getStatus());
                return;
            }
            //反射调用方法，传递logId
            log.info("db消息队列执行器-处理中消息消费,id:{}---消息:{}", queueLog.getId(), queueLog.getMessage());
            R r = invokeConsumer(consumeDo.getClassName(), consumeDo.getMethodName(), () -> {
                MessageCenterQueueMessageDo messageDo = new MessageCenterQueueMessageDo();
                messageDo.setLogId(queueLog.getId());
                messageDo.setMessage(queueLog.getMessage());
                return messageDo;
            });
            rs.add(r);
            log.info("db消息队列执行器-处理中消息消费,id:{}---结果:{}", queueLog.getId(), JSON.toJSONString(r));
        });
        return R.success(rs);
    }

}
